use std::{str::FromStr, time::Duration};
use tracing::info;
use tracing::Level;
mod config;
mod db;
mod feed;
mod post;

use anyhow::{Context, Result};
use chrono::{DateTime, FixedOffset, Utc};
use config::Config;
use db::Db;
use feed::{AtomOrRss, Feed};
use matrix_sdk::config::SyncSettings;
use matrix_sdk::matrix_auth::MatrixSession;
use matrix_sdk::RoomState;
use post::Post;
use regex::Regex;
use std::sync::Arc;
use tokio::io::BufReader;
use tokio::sync::Mutex;
use uuid::Uuid;
// use opml::{Outline,OPML};
// pub fn parse_rss(data: String) {
//     let parsed = OPML::new(&data).unwrap();
//     info!("{:#?}", parsed)
// }
// use syndication::Feed::{self, Atom, RSS};
use atom_syndication::Feed as AtomFeed;
use matrix_sdk::{
    event_handler::Ctx,
    room::Room,
    ruma::{
        api::client::message,
        events::room::message::{
            MessageType, OriginalSyncRoomMessageEvent, RoomMessageEventContent,
        },
        OwnedRoomId, TransactionId,
    },
    Client,
};
use rss::Channel;
use std::env;
use tokio_stream::StreamExt;
static REGEX: OnceLock<Regex> = OnceLock::new();
// use time::{OffsetDateTime, format_description::well_known::Rfc2822};
#[tokio::main(flavor = "multi_thread")]
async fn main() {
    run().await;
}
async fn run() {
    // let mut config;
    tracing_subscriber::fmt().with_max_level(Level::INFO).init();
    let homeserver_url = env::var("HOMESERVER_URL").unwrap();
    let client = Client::builder()
        // .homeserver_url(homeserver_url)
        .sqlite_store("./data/rss", None)
        .homeserver_url(&homeserver_url)
        .build()
        .await
        .unwrap();
    let config = get_config(client.clone()).await;
    let db = Db::connect(&config.lock().await.postgres_path)
        .await
        .unwrap();
    // dbg!(request);
    // let url = "https://stallman.org/rss/rss.xml";
    // let config_clone = config.clone();
    let client_clone = client.clone();
    let client_clone2 = client.clone();
    // let token = config.t
    client.add_event_handler_context(config.clone());
    client.add_event_handler_context(db.clone());
    client.add_event_handler(messages_handler);
    //     tokio::spawn(async move {
    //         client.sync(SyncSettings::default()).await.unwrap();
    //     info!("started!");
    // }).await.unwrap();
    // let _ = tokio::spawn(async move { start_update(client_clone, config, db).await }).await;
    // info!("test2");
    // let _ = tokio::spawn(async move { start_sync(client_clone2).await }).await;
    let _ = tokio::join!(
        start_update(client_clone, config, db),
        start_sync(client_clone2)
    );
    // client.sync(SyncSettings::default()).await.unwrap();
    // tokio::spawn( async move {
    // }).await.unwrap();
}
use std::sync::OnceLock;
async fn parse_feed_first(url: &str, db: Db) -> Result<()> {
    let data = reqwest::get(url).await?.text().await?;
    let last_update_time = Utc::now();
    info!("parsing new feed");
    // let mut buf_data = BufReader::new(data);
    // let cell = OnceLock::new();
    REGEX.get_or_init(|| Regex::new(r"(?i)<\s*feed[^>]*>").unwrap());
    // .as_ref()
    // .unwrap();
    if REGEX.get().unwrap().is_match(&data) {
        // let data = data.as_bytes();
        let mut reader = BufReader::new(data.as_bytes());
        if AtomFeed::read_from(&mut reader).await.is_ok() {
            let feed = Feed::new(
                Uuid::new_v4(),
                /*update_time*/ last_update_time,
                url.to_string(),
                AtomOrRss::Atom,
            );
            db.add_feed(feed)
                .await
                .context("Error during adding feed to db:")?;
        }
    } else if Channel::read_from(BufReader::new(data.as_bytes())).await.is_ok() {
        // info!("rss feed");
        let feed = Feed::new(
            Uuid::new_v4(),
            /*update_time*/ last_update_time,
            url.to_string(),
            AtomOrRss::Rss,
        );
        db.add_feed(feed).await?;
        // parse_channel(channel, update_time);
    }
    Ok(())
}
async fn parse_feed(feed: &Feed, update_time: DateTime<Utc>) -> Result<Vec<Post>> {
    let data = reqwest::get(&feed.url).await?.text().await?;
    let mut buf_data = BufReader::new(data.as_bytes());
    // let update_time = Utc::now();
    // if let Ok(atom_feed) = AtomFeed::from_str(&data){
    match feed.feed_type {
        AtomOrRss::Atom => {
            let atom_feed = AtomFeed::read_from(&mut buf_data).await?;
            info!("atom feed!");
            parse_atom_feed(atom_feed, update_time, feed.last_update_time)
        }
        AtomOrRss::Rss => {
            info!("rss feed");
            let channel = Channel::read_from(&mut buf_data).await?;
            parse_channel(channel, update_time, feed.last_update_time)
        }
    }
    // feed.last_update_time =
}
fn parse_channel(
    channel: Channel,
    update_time: DateTime<Utc>,
    last_update_time: DateTime<Utc>,
) -> Result<Vec<Post>> {
    let mut posts = Vec::new();
    for item in channel.items {
        // info!("{:#?}", item.pub_date());
        let parsed_date = DateTime::parse_from_rfc2822(item.pub_date().context("item pub_date does not exists")?)
            // .ok()
            ?;
        let update_time: DateTime<FixedOffset> = update_time.into();
        // info!("{} {}", parsed_date, update_time);
        // match parsed_date.cmp(&update_time){
        //     std::cmp::Ordering::Equal => info!("Equal!"),
        //     std::cmp::Ordering::Greater => info!("parsed_date > update_time"),
        //     std::cmp::Ordering::Less => info!("parsed_date < update_time")
        // }
        if parsed_date < update_time && parsed_date >= last_update_time {
            let post = Post::new(item.title, item.link.context("item link does not exists")?);
            posts.push(post)
        }
        posts.reverse();
    }
    //     let posts = channel
    //     .items
    //     .iter()
    //     .map(|item| info!("{item:?}"))
    // // unimplemented!();
    //     .filter(|item| {
    //         let parsed_date = DateTime::parse_from_rfc2822(item.pub_date().unwrap())
    //             .ok()
    //             .unwrap();
    //         parsed_date > update_time //&& parsed_date <= peek_time
    //     })
    //     .map(|item| Post::new( item.title().map(|x| x.to_owned()),item.link().unwrap().to_string()))
    //     .collect::<Vec<_>>();
    Ok(posts)
}
fn parse_atom_feed(
    feed: AtomFeed,
    update_time: DateTime<Utc>,
    last_update_time: DateTime<Utc>,
) -> Result<Vec<Post>> {
    // info!("parsing atom feed: {:#?}", feed);
    let mut posts = feed
        .entries()
        .iter()
        .filter(|item| {
            let parsed_date = item.updated();
            parsed_date < &update_time && parsed_date >= &last_update_time
        })
        .map(|item| {
            info!("item: {:#?}", item.links);
            Post::new(
                Some(item.title().value.clone()),
                item.links()
                    .first()
                    .context("error getting atom entry Links")
                    .unwrap()
                    .href()
                    .to_string(),
            )
        })
        .collect::<Vec<_>>();
    posts.reverse();
    Ok(posts)
}
pub type FixedDateTime = DateTime<FixedOffset>;
async fn messages_handler(
    event: OriginalSyncRoomMessageEvent,
    room: Room,
    // client: Client,
    // config: Ctx<Arc<Mutex<Config>>>,
    db: Ctx<Db>,
) -> Result<()> {
    info!("new message");
    // let config = config.0.clone();
    if let RoomState::Joined = room.state() {
        if let MessageType::Text(message) = event.content.msgtype.clone() {
            let text = {
                if let Some(formatted_message) = message.formatted {
                    formatted_message.body
                } else {
                    message.body
                }
            };
            if text.starts_with("!add_feed") {
                let url = &text.strip_prefix("!add_feed ").unwrap();
                let feed = parse_feed_first(url, db.clone()).await;
                match feed {
                    Ok(()) => {
                        let content =
                            RoomMessageEventContent::text_plain(format!("Added url: {url}"));
                        room.send(content).await?;
                    }
                    Err(e) => {
                        let content = RoomMessageEventContent::text_plain(format!(
                            "Error while adding url: {e}"
                        ));
                        room.send(content).await?;
                    }
                }
                // info!(feed)
            }
            if text.starts_with("!list_feeds") {
                let mut answer_text = String::from("Feeds:\n");
                for feed in db.get_feeds().await.context("Error in !list_feeds")? {
                    answer_text.push_str(&format!("{}\n", &feed.url()));
                }
                let content = RoomMessageEventContent::text_plain(answer_text);
                room.send(content).await?;
            }
        }
    }
    Ok(())
}
pub async fn get_config(client: Client) -> Arc<Mutex<Config>> {
    if let Ok(path) = env::var("CONFIG_PATH") {
        let config = Arc::new(Mutex::new(Config::load_config(&path).await.unwrap()));
        let config_clone = config.clone();
        let config_lock = config_clone.lock().await;
        let session = MatrixSession {
            meta: matrix_sdk::SessionMeta {
                user_id: config_lock.user_id.clone().try_into().unwrap(),
                device_id: config_lock.device_id.clone().into(),
            },
            tokens: matrix_sdk::matrix_auth::MatrixSessionTokens {
                access_token: config_lock.access_token.clone(),
                refresh_token: None,
            },
        };
        client.matrix_auth().restore_session(session).await.unwrap();
        config
    } else {
        info!("use default config path!");
        let path = "./data/config.json".to_string();
        let user_name = env::var("USERNAME").unwrap();
        let password = env::var("PASSWORD").unwrap();
        let room_id = env::var("ROOM_ID").unwrap();
        let homeserver_url = env::var("HOMESERVER_URL").unwrap();
        let postgres_path = env::var("postgres_path").unwrap();
        // config = Arc::new(Mutex::new(Config::new(user_name, password, homeserver_url, room_id, path)));
        // config.lock().await.save_config().await;
        let session = client
            .matrix_auth()
            .login_username(&user_name, &password)
            .initial_device_display_name("rss_bot")
            .await
            .unwrap();
        let config = Config::new(
            session.user_id.to_string(),
            session.device_id.to_string(),
            session.access_token,
            homeserver_url,
            room_id,
            path,
            postgres_path,
        );
        config.save_config().await;
        Arc::new(Mutex::new(config))
    }
}
pub async fn start_update(client: Client, config: ConfigWrapper, db: Db) -> Result<()> {
    info!("Start update");
    // let mut update_time = Utc.with_ymd_and_hms(1990, 01, 01, 00, 00, 00).single().unwrap();
    let mut update_time = Utc::now();
    let config_clone = config.clone();
    let config_lock = config_clone.lock().await;
    let room_id = config_lock.room_id.clone();
    drop(config_lock);
    // drop()
    loop {
        info!("new loop!");
        let config = config.lock().await;
        drop(config);
        let mut feeds = tokio_stream::iter(
            db.get_feeds()
                .await
                .context("Error(getting feeds in a loop)")?,
        );
        // info!("feeds: {:#?}", config.feeds);
        // for feed in &mut config.feeds {

        while let Some(feed) = feeds.next().await {
            let client_clone = client.clone();
            let db_clone = db.clone();
            let room_id_clone = room_id.clone();
            info!("test");
            let _ = tokio::spawn(async move {
                start_parse(feed, room_id_clone, client_clone, db_clone, update_time).await
            })
            .await;
            // }
        }
        tokio::time::sleep(Duration::from_secs(60 * 15)).await;
        update_time = Utc::now();
    }
}
type ConfigWrapper = Arc<Mutex<Config>>;

async fn start_parse(
    feed: Feed,
    room_id: String,
    client: Client,
    db: Db,
    update_time: DateTime<Utc>,
) -> Result<()> {
    info!("parsing feed: {}", feed.id);

    if let Ok(posts) = parse_feed(&feed, update_time).await {
        let mut posts = tokio_stream::iter(posts);
        // posts.
        while let Some(post) = posts.next().await {
            let body = format!(
                "New post: {title} \n {url}",
                title = post.title.clone().unwrap_or("New post:".to_string()),
                url = post.url()
            );
            let content = RoomMessageEventContent::text_plain(body);
            let room_id = OwnedRoomId::from_str(&room_id).unwrap();
            let request = message::send_message_event::v3::Request::new(
                room_id,
                TransactionId::new(),
                &content,
            )
            .unwrap();
            client.send(request, None).await.unwrap();
            // tokio::time::sleep(Duration::from_secs(15)).await;
            // update_time = Utc::now();
        }
    }
    db.update_update_time(update_time, feed.id)
        .await
        .context("Error during updating tiem")?;
    Ok(())
}

async fn start_sync(client: Client) -> Result<()> {
    info!("Start sync!");
    let sync_settings = SyncSettings::new()/* .timeout(Duration::from_secs(15))*/;
    client.sync(sync_settings).await?;
    Ok(())
}
